package com.imooc.hadoop.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URI;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Hadoop HDFS Java API 操作
 */
public class HDFSApp {

    public static final String HDFS_PATH = "hdfs://10.1.0.52:8020";

    FileSystem fileSystem = null;
    //默认读取classPath下的XXX-site.xml(core-site.xml,...)配置文件，并解析其内容，封装到configuration对象中
    //也可以在代码中进行设置
    Configuration configuration = null;
    @Before
    public void setUp() throws Exception {
        System.out.println("HDFSApp.setUp");
        configuration = new Configuration();
        fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, "hadoop");//hadoop用户
//        configuration.set("fs.defaultFS","hdfs://10.1.0.52:8020");
//        fileSystem = fileSystem.get(configuration);
    }

    @After
    public void tearDown() throws Exception {
        configuration = null;
        fileSystem = null;

        System.out.println("HDFSApp.tearDown");
    }

    /**
     * 创建HDFS目录
     */
    @Test
    public void mkdir() throws Exception {
        fileSystem.mkdirs(new Path("/hdfsapi/test/tt"));
        fileSystem.mkdirs(new Path("/hdfsapi/test/permission"), new FsPermission("111"));
    }

    /**
     * 创建文件
     */
    @Test
    public void create() throws Exception {
        FSDataOutputStream output = fileSystem.create(new Path("/hdfsapi/test/a.txt"));
        output.write("hello hadoop".getBytes()); //创建一个a.txt文件并写入hello hadoop
        output.flush();
        output.close();
    }

    /**
     * 查看HDFS文件的内容
     */
    @Test
    public void cat() throws Exception {
        FSDataInputStream in = fileSystem.open(new Path("/hdfsapi/test/a.txt"));
        IOUtils.copyBytes(in, System.out, 1024);
        in.close();
    }


    /**
     * 重命名
     */
    @Test
    public void rename() throws Exception {
        Path oldPath = new Path("/hdfsapi/test/a.txt");
        Path newPath = new Path("/hdfsapi/test/b.txt");
        fileSystem.rename(oldPath, newPath);
    }

    /**
     * 上传文件到HDFS
     *
     * @throws Exception
     */
    @Test
    public void copyFromLocalFile() throws Exception {
        Path localPath = new Path("F:\\云南业务商户展示属性2019-01-07.txt");
        Path hdfsPath = new Path("/hdfsapi/test");
        fileSystem.copyFromLocalFile(localPath, hdfsPath);
    }

    /**
     * 上传文件到HDFS
     */
    @Test
    public void copyFromLocalFileWithProgress() throws Exception {
        InputStream in = new BufferedInputStream(
                new FileInputStream(
                        new File("F:\\peiqi.mp4")));

        FSDataOutputStream output = fileSystem.create(new Path("/hdfsapi/test/peiqi.mp4"),
                new Progressable() {
                    public void progress() {
                        System.out.print(".");  //带进度提醒信息
                    }
                });

        IOUtils.copyBytes(in, output, 4096);
    }

    /**
     * 自己控制文件上传
     *
     * @throws Exception
     */
    @Test
    public void copyFromLocalFileWithProgress2() throws Exception {
        InputStream in = new BufferedInputStream(
                new FileInputStream(
                        new File("F:\\peiqi.mp4")));

        final AtomicInteger writeBytes = new AtomicInteger(0);
        FSDataOutputStream output = fileSystem.create(new Path("/hdfsapi/test/peiqi.mp4"),
                new Progressable() {
                    public void progress() {
                        System.out.println("writeBytes = " + writeBytes.get());  //显示当前写入数据数
                    }
                });

        byte[] buffer = new byte[1024];
        int readBytes = in.read(buffer);
        while (readBytes != -1) {
            output.write(buffer);
            output.flush();
            output.hsync();//保证当前实际上写入成功，以后写入失败不影响当前
            writeBytes.addAndGet(readBytes); //记录当前写入数据数 在Progressable中显示
            readBytes = in.read(buffer);
        }
    }


    /**
     * 下载HDFS文件
     */
    @Test
    public void copyToLocalFile() throws Exception {
        Path localPath = new Path("F:\\hadoop\\test\\h.txt");
        Path hdfsPath = new Path("/hdfsapi/test/b.txt");
//        fileSystem.copyToLocalFile(hdfsPath,localPath);
        /**
         通过java操作
         使用public void copyToLocalFile(boolean delSrc, Path src, Path dst, boolean useRawLocalFileSystem)
           1)第一个参数代表是否删除源文件
           2)最后一个参数代表是否使用本地的文件系统（java的文件系统）
         *
         */
        fileSystem.copyToLocalFile(false, hdfsPath, localPath, true);
    }

    /**
     * 查看某个目录下的所有文件
     */
    @Test
    public void listFiles() throws Exception {
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/hdfsapi/test"));
        for (FileStatus fileStatus : fileStatuses) {
            String isDir = fileStatus.isDirectory() ? "文件夹" : "文件";
            short replication = fileStatus.getReplication();
            long len = fileStatus.getLen();
            String path = fileStatus.getPath().toString();

            System.out.println(isDir + "\t" + replication + "\t" + len + "\t" + path);
        }

    }

    /**
     * 删除
     */
    @Test
    public void delete() throws Exception {
        fileSystem.delete(new Path("/hdfsapi/test/hadoop-2.6.0-cdh5.7.0.tar.gz"), true); //递归删除
    }


}
